package com.wngbms.redisson;

import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 演示redisson延时队列
 */
@Slf4j
@Component
public class RedissonQueue {

    @Autowired
    RedissonClient redissonClient;

    @Bean
    public RBlockingQueue<String> blockingQueue() {
        return redissonClient.getBlockingQueue("bqdemo");
    }

    @Bean
    public RDelayedQueue<String> delayedQueue(RBlockingQueue<String> blockingQueue) {
        return redissonClient.getDelayedQueue(blockingQueue);
    }

    ExecutorService executorService = Executors.newFixedThreadPool(1);

    /**
     * 消费者
     *
     * @param blockingQueue
     * @return
     */
    @Bean
    public ApplicationRunner run1(RBlockingQueue<String> blockingQueue) {
        return args -> {
            for (int i = 0; i < 1; i++) {
                executorService.submit(() -> {
                    while (true) {
                        String take = null;
                        try {
                            take = blockingQueue.take();
                        } catch (InterruptedException e) {
                            log.error("{}", e.getMessage(), e);
                        }
                        log.info("take:[{}]", take);
                    }
                });
            }
        };
    }

    /**
     * 生产者
     *
     * @param delayedQueue
     * @return
     */
//    @Bean
    public ApplicationRunner run2(RDelayedQueue<String> delayedQueue) {
        return args -> {

            new Thread(() -> {

                while (true) {
                    String uuid = UUID.randomUUID().toString();
                    log.info("offer:[{}]", uuid);
                    delayedQueue.offer(uuid, 30, TimeUnit.SECONDS);
                    try {
                        Thread.sleep(15000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, "run2").start();
        };
    }

}
